package tableapi

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table, TableEnvironment}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, OldCsv, Schema}
import org.apache.flink.table.types.DataType

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object Table3_Api {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env)

    //连接外部File系统，读取数据，注册表
    tableEnv.connect(new FileSystem()
      .path(inputPath)
    )
      //      .withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("fileTable")

    val fileTable: Table = tableEnv.from("fileTable")

    fileTable.toAppendStream[(String, Long, Double)].print("file")

    tableEnv.connect(new Kafka()
      .version("0.11")
      .topic("sensor")
      .property("zookeeper.connect", "192.168.88.106:2181")
      .property("bootstrap.servers", "192.168.88.106:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("kafkaTable")
    val kafkaTable: Table = tableEnv.from("kafkaTable")

    kafkaTable.toAppendStream[(String, Long, Double)].print("kafka")

    //查询转换
    //使用table api
    val resultTable: Table = fileTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")
    resultTable.toAppendStream[(String,Double)].print("table")

    //Sql
    val resltSqlTable:Table = tableEnv.sqlQuery(
      """
        |select id,temperature
        | from fileTable
        |  where id = 'sensor_1'
      """.stripMargin)
    resltSqlTable.toAppendStream[(String,Double)].print("sql")

    env.execute("table api test")
  }


}
